package com.gome.transactiontree.repository;

import java.util.List;

import org.apache.commons.lang.StringUtils;

import redis.Gcache;
import redis.GcacheClient;
import redis.gcache.KeysExecutor;

import com.gome.transactiontree.api.Transaction;
import com.gome.transactiontree.api.TransactionCallback;
import com.gome.transactiontree.logging.TransactionLogger;
import com.gome.transactiontree.serializer.KryoThreadLocalSerializer;
import com.gome.transactiontree.serializer.ObjectSerializer;

public class RedisTransactionRepository implements TransactionStore {

	public static int EXPIRE_TIME = 24 * 60 * 60;
	private ObjectSerializer<Transaction> serializer = new KryoThreadLocalSerializer<Transaction>();
	
	private Gcache gcache;
	private String zookeeperAddress;
	private String business;

	@Override
	public Transaction get(String transactionId) {
		String fullKey = getFullFileName(transactionId);
		return serializer.deserialize(gcache.getBytes(fullKey));
	}

	@Override
	public void put(String transactionId, Transaction transaction) {
		String fullKey = getFullFileName(transactionId);
		gcache.setex(fullKey, EXPIRE_TIME, serializer.serialize(transaction));
	}

	@Override
	public void delete(String transactionId) {
		String fullKey = getFullFileName(transactionId);
		gcache.del(fullKey);
	}

	@Override
	public void foreach(final TransactionCallback<Object> action) {
		gcache.scanCluster(business + "*", 1000, new KeysExecutor() {
	        @Override
	        public void execute(List<String> keys) {
	            if (!keys.isEmpty()) {
	            	for(String key :keys){
	        			Transaction transaction = serializer.deserialize(gcache.getBytes(key));
	        			try {
	        				action.doInTransaction(transaction);
	        			} catch (Throwable e) {
	        				throw new IllegalStateException(e.getMessage(), e);
	        			}
	                }
	            }
	        }
	    });
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		if(gcache == null){
			if(StringUtils.isEmpty(zookeeperAddress)){
				throw new IllegalStateException("please check your zookeeperAddress, maybe it zookeeperAddress empty!");
			}
			if(StringUtils.isEmpty(business)){
				throw new IllegalStateException("please check your business, maybe it business empty!");
			}
			gcache = new GcacheClient(zookeeperAddress, business);
			
			TransactionLogger.REPOSITORY.info("初始化gcache存储成功，业务key: {}", business);
		}
	}

	@Override
	public void destroy() throws Exception {
		gcache.close();
		
		TransactionLogger.REPOSITORY.info("关闭gcache存储成功");
	}
	
	private String getFullFileName(String transactionId) {
        return String.format("%s/%s", business, transactionId);
    }

	public String getZookeeperAddress() {
		return zookeeperAddress;
	}

	public void setZookeeperAddress(String zookeeperAddress) {
		this.zookeeperAddress = zookeeperAddress;
	}

	public String getBusiness() {
		return business;
	}

	public void setBusiness(String business) {
		this.business = business;
	}
}
